package org.budo.warehouse.logic.consumer.jdbc;

import java.util.ArrayList;
import java.util.List;

import org.budo.graph.annotation.SpringGraph;
import org.budo.support.lang.util.StringUtil;
import org.budo.warehouse.logic.api.AbstractDataConsumer;
import org.budo.warehouse.logic.api.DataEntry;
import org.budo.warehouse.logic.api.DataEntry.EventType;
import org.budo.warehouse.logic.api.DataMessage;
import org.budo.warehouse.logic.util.DataEntryUtil;
import org.budo.warehouse.logic.util.PipelineUtil;

import lombok.extern.slf4j.Slf4j;

/**
 * @author limingwei
 */
@Slf4j
public abstract class AbstractJdbcDataConsumer extends AbstractDataConsumer {
    /**
     * @see com.alibaba.otter.canal.protocol.CanalEntry.EventType
     */
    @SpringGraph
    public List<SqlUnit> buildSql(DataMessage dataMessage) {
        List<SqlUnit> sqlUnits = new ArrayList<SqlUnit>();

        List<DataEntry> dataEntries = dataMessage.getDataEntries();
        for (DataEntry dataEntry : dataEntries) {
            String eventType = dataEntry.getEventType();

            if (EventType.INSERT.equalsIgnoreCase(eventType)) {
                List<SqlUnit> insertSql = this.dataEntryInsertSql(dataEntry);
                sqlUnits.addAll(insertSql);
                continue;
            }

            if (EventType.DELETE.equalsIgnoreCase(eventType)) {
                List<SqlUnit> deleteSql = this.dataEntryDeleteSql(dataEntry);
                sqlUnits.addAll(deleteSql);
                continue;
            }

            if (EventType.UPDATE.equalsIgnoreCase(eventType)) {
                List<SqlUnit> updateSql = this.dataEntryUpdateSql(dataEntry);
                sqlUnits.addAll(updateSql);
                continue;
            }

            // 这里没有处理表名表达式 就是说 不支持源表和目标表名称不同
            if (DataEntryUtil.isDDL(dataEntry) //
                    && !DataEntryUtil.isDropPartition(dataEntry)) { // 暂不支持 分区表操作 DROP PARTITION
                String sql = "USE " + dataEntry.getSchemaName() + "; " // 切换数据库
                        + dataEntry.getSql(); // 执行DDL
                sqlUnits.add(new SqlUnit(sql, null));
                continue;
            }

            // 未支持的语句 非 DDL 非 DML, TRUNCATE RENAME
            log.info("#47 eventType=" + eventType + ", dataEntry=" + dataEntry + ", pipeline=" + this.getPipeline());
        }

        return sqlUnits;
    }

    @SpringGraph
    protected List<SqlUnit> dataEntryInsertSql(DataEntry dataEntry) {
        List<SqlUnit> sqlUnits = new ArrayList<SqlUnit>();

        Integer rowCount = dataEntry.getRowCount();
        for (int rowIndex = 0; rowIndex < rowCount; rowIndex++) {
            SqlUnit insertRow = this.dataRowInsertSql(dataEntry, rowIndex);
            sqlUnits.add(insertRow);
        }

        return sqlUnits;
    }

    @SpringGraph
    protected List<SqlUnit> dataEntryDeleteSql(DataEntry dataEntry) {
        List<SqlUnit> sqlUnits = new ArrayList<SqlUnit>();

        Integer rowCount = dataEntry.getRowCount();
        for (int rowIndex = 0; rowIndex < rowCount; rowIndex++) {
            SqlUnit deleteRow = this.dataRowDeleteSql(dataEntry, rowIndex);
            sqlUnits.add(deleteRow);
        }

        return sqlUnits;
    }

    public List<SqlUnit> dataEntryUpdateSql(DataEntry dataEntry) {
        List<SqlUnit> sqlUnits = new ArrayList<SqlUnit>();

        Integer rowCount = dataEntry.getRowCount();
        for (int rowIndex = 0; rowIndex < rowCount; rowIndex++) {
            SqlUnit updateRow = this.dataRowUpdateSql(dataEntry, rowIndex);
            sqlUnits.add(updateRow);
        }

        return sqlUnits;
    }

    /**
     * 拼装 insert 语句
     */
    public SqlUnit dataRowInsertSql(DataEntry dataEntry, int rowIndex) {
        List<Object> parameters = new ArrayList<Object>();

        List<String> fields = new ArrayList<String>();
        List<String> values = new ArrayList<String>();

        Integer columnCount = dataEntry.getColumnCount(rowIndex);
        for (int columnIndex = 0; columnIndex < columnCount; columnIndex++) {
            String columnName = dataEntry.getColumnName(rowIndex, columnIndex);
            fields.add("`" + columnName + "`");

            Object columnValueAfter = dataEntry.getColumnValueAfter(rowIndex, columnIndex);
            if (null == columnValueAfter) {
                values.add("NULL");
                continue;
            }

            values.add("?");
            parameters.add(columnValueAfter);
        }

        String table = PipelineUtil.targetSchemaTable(this.getPipeline(), dataEntry);
        String sql = "INSERT INTO " + table + "(" + StringUtil.join(fields, ", ") + ") VALUES (" + StringUtil.join(values, ", ") + ") ";
        return new SqlUnit(sql, parameters.toArray());
    }

    public SqlUnit dataRowDeleteSql(DataEntry dataEntry, int rowIndex) {
        List<Object> parameters = new ArrayList<Object>();
        List<String> where = this.whereSql(dataEntry, rowIndex, parameters);

        String table = PipelineUtil.targetSchemaTable(this.getPipeline(), dataEntry);
        String sql = "DELETE FROM " + table + " WHERE " + StringUtil.join(where, " AND ");
        return new SqlUnit(sql, parameters.toArray());
    }

    public SqlUnit dataRowUpdateSql(DataEntry dataEntry, int rowIndex) {
        List<Object> parameters = new ArrayList<Object>();

        List<String> set = this.dataRowUpdateSetSql(dataEntry, rowIndex, parameters);
        List<String> where = this.whereSql(dataEntry, rowIndex, parameters);

        String table = PipelineUtil.targetSchemaTable(this.getPipeline(), dataEntry);
        String sql = "UPDATE " + table + " SET " + StringUtil.join(set, ", ") + " WHERE " + StringUtil.join(where, " AND ");
        return new SqlUnit(sql, parameters.toArray());
    }

    public List<String> dataRowUpdateSetSql(DataEntry dataEntry, int rowIndex, List<Object> parameters) {
        List<String> set = new ArrayList<String>();

        Integer columnCount = dataEntry.getColumnCount(rowIndex);
        for (int columnIndex = 0; columnIndex < columnCount; columnIndex++) {
            String columnName = dataEntry.getColumnName(rowIndex, columnIndex);
            Object columnValueAfter = dataEntry.getColumnValueAfter(rowIndex, columnIndex);
            if (null == columnValueAfter) {
                set.add("`" + columnName + "`=NULL"); // 空值
                continue;
            }

            set.add("`" + columnName + "`=?");
            parameters.add(columnValueAfter);
        }

        return set;
    }

    public List<String> whereSql(DataEntry dataEntry, int rowIndex, List<Object> parameters) {
        List<String> where = new ArrayList<String>();
        boolean hasIsKeyColumn = DataEntryUtil.hasIsKeyColumn(dataEntry, rowIndex);
        Integer columnCount = dataEntry.getColumnCount(rowIndex);

        for (int columnIndex = 0; columnIndex < columnCount; columnIndex++) {
            if (hasIsKeyColumn && !dataEntry.getColumnIsKey(rowIndex, columnIndex)) { // 如果有主键就主键作为条件,否则全部作为条件
                continue;
            }

            String columnName = dataEntry.getColumnName(rowIndex, columnIndex);
            Object columnValueBefore = dataEntry.getColumnValueBefore(rowIndex, columnIndex);
            if (null == columnValueBefore) {
                where.add("`" + columnName + "` IS NULL"); // 空值
                continue;
            }

            where.add("`" + columnName + "`=?");
            parameters.add(columnValueBefore);
        }

        return where;
    }
}